/*
 * IoService.cpp
 *
 *  Created on: Nov 4, 2012
 *      Author: work
 */

#include "IoService.h"
#include <sys/epoll.h>

namespace zeus
{

IoService::IoService() :
		_is_running(true), _epoll_fd(::epoll_create(5000))
{

}

IoService::~IoService()
{
}

void IoService::loop()
{
	while (_is_running == true)
	{
		int32 ms = 10;

		poll(ms);
	}
}

void IoService::quit()
{
	_is_running = false;
}

void IoService::wake()
{

}

void IoService::post(const EmptyHandler & handler)
{

}

void IoService::read_op(int32 fd, Operation * op)
{
	Channel & channel = _channels[fd];

	if (channel._epoll_add == false)
	{
		channel.set_fd(fd);

		epoll_add(channel,EPOLLIN);

		channel._epoll_add = true;
		channel._epoll_read = true;
	}

	if (channel._epoll_read == false)
	{
		epoll_mod(channel,EPOLLIN);

		channel._epoll_read = true;
	}

	_read_ops.add_opertion(fd, op);
}

void IoService::write_op(int32 fd, Operation * op)
{
	Channel & channel = _channels[fd];

	if (channel._epoll_add == false)
	{
		epoll_add(channel,EPOLLOUT);

		channel._epoll_add = true;

		channel._epoll_write = true;
	}

	if (channel._epoll_write == false)
	{
		epoll_mod(channel,EPOLLOUT);
		channel._epoll_write = true;
	}

	_write_ops.add_opertion(fd, op);
}

void IoService::epoll_add(Channel & channel,int32 event)
{
	epoll_event ev =
	{ 0,
	{ 0 } };

	ev.events = EPOLLERR | EPOLLHUP | event;

	ev.data.fd = channel.get_fd();

	int result = ::epoll_ctl(_epoll_fd, EPOLL_CTL_MOD, channel.get_fd(), &ev);
	if (result != 0 && errno == ENOENT)
		result = ::epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, channel.get_fd(), &ev);
}

void IoService::epoll_mod(Channel & channel,int32 event)
{
	epoll_event ev =
	{ 0,
	{ 0 } };
	ev.events = EPOLLERR | EPOLLHUP | event;

	ev.data.fd = channel.get_fd();

	if (channel._epoll_read == true)
	{
		ev.events = EPOLLERR | EPOLLHUP | EPOLLIN;
	}

	if (channel._epoll_write == true)
	{
		ev.events = EPOLLERR | EPOLLHUP | EPOLLOUT;
	}

	int result = ::epoll_ctl(_epoll_fd, EPOLL_CTL_MOD, channel.get_fd(), &ev);
	if (result != 0 && errno == ENOENT)
		result = ::epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, channel.get_fd(), &ev);
}

void IoService::poll(int32 wait_ms)
{
	epoll_event events[128];

	int32 num_events = ::epoll_wait(_epoll_fd, events, 128, wait_ms);

	for (int32 i = 0; i < num_events; ++i)
	{
		int32 fd = events[i].data.fd;

		bool more_reads = false;
		bool more_writes = false;
		bool more_except = false;

		Error ec;

		if (events[i].events & (EPOLLPRI | EPOLLERR | EPOLLHUP))	//except
		{

		}

		if (events[i].events & (EPOLLIN | EPOLLERR | EPOLLHUP))
		{
			more_reads = _read_ops.perform(fd, ec);
		}
		else
		{
			more_reads = _read_ops.has_operation(fd);
		}

		if (events[i].events & (EPOLLOUT | EPOLLERR | EPOLLHUP))
		{
			more_writes = _write_ops.perform(fd, ec);
		}
		else
		{
			more_writes = _write_ops.has_operation(fd);
		}

		if ((events[i].events & (EPOLLERR | EPOLLHUP)) != 0
				&& (events[i].events & ~(EPOLLERR | EPOLLHUP)) == 0
				&& !more_except && !more_reads && !more_writes)
		{
			// If we have an event and no operations associated with the
			// descriptor then we need to delete the descriptor from epoll. The
			// epoll_wait system call can produce EPOLLHUP or EPOLLERR events
			// when there is no operation pending, so if we do not remove the
			// descriptor we can end up in a tight loop of repeated
			// calls to epoll_wait.
			epoll_event ev =
			{ 0,
			{ 0 } };
			::epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, &ev);
		}
		else
		{
			epoll_event ev =
			{ 0,
			{ 0 } };
			ev.events = EPOLLERR | EPOLLHUP;
			if (more_reads)
				ev.events |= EPOLLIN;
			if (more_writes)
				ev.events |= EPOLLOUT;
			if (more_except)
				ev.events |= EPOLLPRI;
			ev.data.fd = fd;

			int result = ::epoll_ctl(_epoll_fd, EPOLL_CTL_MOD, fd, &ev);
			if (result != 0 && errno == ENOENT)
				result = ::epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, fd, &ev);
			if (result != 0)
			{
				ec = Error(1);

				_read_ops.perform_all_operations(fd, ec);
				_write_ops.perform_all_operations(fd, ec);
			}
		}

	}
	_read_ops.complete();

	_write_ops.complete();

}

} /* namespace zeus */
